package com.study.bigdata.spark.core.req

import org.apache.spark.{SparkConf, SparkContext}

object Spark01_Req_HotCategoryTop10_2 {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")
    val sc = new SparkContext(conf)
    // TODO 需求一：Top10热门品类
    // TODO 问题：1.同一个RDD的重复使用 2.cogroup算子可能出现笛卡尔乘积，性能低下

    // TODO 读取文件，获取原始数据
    val fileDatas = sc.textFile("data/user_visit_action.txt")
    val flatDatas = fileDatas.flatMap(
      data => {
        val datas = data.split("_")
        if (datas(6) != "-1") {
          List((datas(6), (1, 0, 0)))
        } else if (datas(8) != "null") {
          val id = datas(8)
          val ids = id.split(",")
          ids.map(
            id => {
              (id, (0, 1, 0))
            }
          )

        } else if (datas(10) != "null") {
          val id = datas(10)
          val ids = id.split(",")
          ids.map(
            id => {
              (id, (0, 0, 1))
            }
          )
        } else {
          Nil
        }
      }
    )
    val top10 = flatDatas.reduceByKey(
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
      }
    ).sortBy(_._2, false).take(10)
    top10.foreach(println)

    sc.stop()

  }

}
